package com.zhouyu.mysql;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.comparator.CompareUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.file.LineSeparator;
import cn.hutool.json.JSONUtil;
import com.zhouyu.mysql.core.*;
import com.zhouyu.mysql.dict.*;
import com.zhouyu.mysql.lock.Lock;
import com.zhouyu.mysql.lock.LockSys;
import com.zhouyu.mysql.lock.RecId;
import com.zhouyu.mysql.page.IndexPage;
import com.zhouyu.mysql.trx.TrxSys;
import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.Function;
import net.sf.jsqlparser.expression.StringValue;
import net.sf.jsqlparser.expression.operators.relational.EqualsTo;
import net.sf.jsqlparser.expression.operators.relational.ExpressionList;
import net.sf.jsqlparser.schema.Column;
import net.sf.jsqlparser.schema.Table;
import net.sf.jsqlparser.statement.alter.Alter;
import net.sf.jsqlparser.statement.alter.AlterExpression;
import net.sf.jsqlparser.statement.create.table.CreateTable;
import net.sf.jsqlparser.statement.create.table.Index;
import net.sf.jsqlparser.statement.insert.Insert;
import net.sf.jsqlparser.statement.select.*;

import java.io.BufferedReader;
import java.io.File;
import java.util.*;

import static com.zhouyu.mysql.page.IndexPage.PAGE_NEW_INFIMUM;

/**
 * 大都督周瑜（我的微信: dadudu6789）
 */
public class SqlExecutor {

    private HaInnodb haInnodb = new HaInnodb();

    public void createTable(CreateTable createTableStatement) {
        haInnodb.createTable(createTableStatement);
    }

    public void insert(Insert insertStatement) {
        haInnodb.insert(insertStatement);
    }

    /**
     * 支持主键索引查询、全表扫描、辅助索引查询、覆盖索引、回表查询
     * @param plainSelectStatement sql语句
     * @param isAll 是不是全表查询
     * @param isSecondary 是不是走的辅助索引
     * @param rootPageNo 走索引时，对应索引的根页页号
     * @return 查询结果
     */
    public String select(PlainSelect plainSelectStatement, boolean isAll, boolean isSecondary, int rootPageNo) {

        // 获取表名
        final String[] tableNameArray = new String[1];
        plainSelectStatement.getFromItem().accept(new FromItemVisitorAdapter() {
            @Override
            public void visit(Table table) {
                tableNameArray[0] = table.getName();
            }
        });
        String tableName = tableNameArray[0];

        DictTable dictTable = SystemDict.getInstance().getNameTables().get(tableName);

        DTuple dTuple = DTupleUtil.convert(plainSelectStatement);

        List<HashMap<String, ? extends Object>> result = select(plainSelectStatement, dTuple, dictTable, isAll, isSecondary, rootPageNo);

        return result.toString();
    }

    /**
     * @param dTuple      走索引时的where条件，如果是全表扫描则不关心
     * @param dictTable   当前表
     * @param isAll       是不是全表扫描
     * @param isSecondary 是不是走的辅助索引
     * @param rootPageNo  走索引时，对应索引的根页页号
     * @return 查询结果
     */
    private List<HashMap<String, ? extends Object>> select(PlainSelect plainSelectStatement, DTuple dTuple, DictTable dictTable, boolean isAll, boolean isSecondary, int rootPageNo) {

        // 收集select字段
        List<String> selectColumnList = new ArrayList<>();
        for (SelectItem<?> selectItem : plainSelectStatement.getSelectItems()) {
            selectColumnList.add(selectItem.toString());
        }

        // 判断是否是order by
        String orderByColumn = null;
        boolean asc = false;
        List<HashMap<String, Object>> sortBuf = new ArrayList<>();
        List<OrderByElement> orderByElements = plainSelectStatement.getOrderByElements();
        if (CollUtil.isNotEmpty(orderByElements)) {
            OrderByElement orderByElement = orderByElements.get(0);
            asc = orderByElement.isAsc();
            Expression expression = orderByElement.getExpression();
            if (expression instanceof Column orderColumn) {
                orderByColumn = orderColumn.getColumnName();
            }
        }

        // 判断是否group by
        String groupByColumn = null;
        String countColumn = null;
        HashMap<String, Integer> heapTable = null;
        GroupByElement groupBy = plainSelectStatement.getGroupBy();
        if (groupBy != null) {
            ExpressionList groupByExpressionList = groupBy.getGroupByExpressionList();
            Object column = groupByExpressionList.get(0);
            if (column instanceof Column groupColumn) {
                groupByColumn = groupColumn.getColumnName();
            }

            SelectItem<?> selectItem = plainSelectStatement.getSelectItem(0);
            Expression expression = selectItem.getExpression();
            if (expression instanceof Function function) {
                if ("count".equals(function.getName())) {
                    Expression parameter = function.getParameters().get(0);
                    if (parameter instanceof Column parameterColumn) {
                        countColumn = parameterColumn.getColumnName();
                    }
                }

            }

            // 如果有group by，就用一张临时表来进行去重，数据量少的时候用内存临时表，数据量多的时候用文件临时表
            heapTable = new HashMap<>();
        }

        PageCur pageCur = new PageCur();
        if (isAll) {
            // 如果是要全表扫描，就取当前表对应的B+树的最左边叶子节点
            pageCur.setPageNo(BtrUtil.getFirstLeafPage(dictTable).fil_page_get_page_offset());
            pageCur.setRec(PAGE_NEW_INFIMUM);
        } else {
            // 走索引，则从对应索引的根页开始
            pageCur.setPageNo(rootPageNo);
            pageCur.setRec(PAGE_NEW_INFIMUM);
        }

        List<HashMap<String, ? extends Object>> result = new ArrayList<>();

        while (true) {
            SelectOneResult selectOneResult = haInnodb.select_one(dictTable, pageCur, dTuple, isAll, isSecondary);
            boolean endOfIndex = selectOneResult.isEndOfIndex();

            if (endOfIndex) {
                break;
            }

            // 如果是全表扫描或走的主键索引，则能直接拿出所有字段，如果走的是辅助索引，则需要进行回表才能拿出所有字段
            HashMap<String, Object> recContent = selectOneResult.getRecContent();
            if (recContent != null) {
                if (isSecondary) {

                    // 如果select的字段在recContent中都有，则可以直接返回，否则需要回表
                    if (CollUtil.isNotEmpty(selectColumnList) && recContent.keySet().containsAll(selectColumnList)) {
                        HashMap<String, Object> resultRecContent = new HashMap<>();
                        for (String selectColumnName : selectColumnList) {
                            resultRecContent.put(selectColumnName, recContent.get(selectColumnName));
                        }
                        result.add(resultRecContent);
                    } else {
                        DTuple idTuple = new DTuple();
                        DField dField = new DField();
                        dField.setDictColumn(DictColumn.builder().name("id").type("int").len(4).build());
                        dField.setData(recContent.get("id"));
                        idTuple.getFields().add(dField);

                        // 回表，就是用id取主键索引上找出结果，所以不是全表扫描，不是辅助索引
                        result.addAll(select(plainSelectStatement, idTuple, dictTable, false, false, dictTable.getPageNo()));
                    }

                } else {
                    if (orderByColumn != null) {
                        sortBuf.add(recContent);
                    } else if (groupByColumn != null) {
                        String groupByValue = (String) recContent.get(groupByColumn);
                        if (heapTable.containsKey(groupByValue)) {
                            Integer count = heapTable.get(groupByValue);
                            if (countColumn == null || recContent.get(countColumn) != null) {
                                ++count;
                            }
                            heapTable.put(groupByValue, count);
                        } else {
                            heapTable.put(groupByValue, (countColumn == null || recContent.get(countColumn) != null) ? 1 : 0);
                        }
                    } else {

                        String forMode = plainSelectStatement.getForMode().getValue();
                        if ("UPDATE".equals(forMode)) {
                            // 加锁逻辑
                            int lock_mode = 3;    // 0000 0000 0000 0011   // enum lock_mode
                            int gap_mode = 1024;  // 0000 0100 0000 0000
                            int mode = lock_mode | gap_mode;

                            int spaceId = dictTable.getSpaceId();
                            int pageNo = selectOneResult.getPageCur().getPageNo();
                            int heapNo = selectOneResult.getPageCur().getHeapNo();
                            RecId recId = new RecId(spaceId, pageNo, heapNo);

                            // 先判断当前行上有没有锁
                            Set<Lock> locks = LockSys.getLock(recId);

                            // 没有加锁，就直接加锁即可
                            if (CollUtil.isEmpty(locks)) {
                                LockSys.addLock(recId, Lock.builder()
                                        .trx(TrxSys.trxThreadLocal.get())
                                        .mode(mode)
                                        .recId(recId)
                                        .lock(new Object())
                                        .thread(Thread.currentThread())
                                        .build());
                            } else {
                                // 当前记录上已经有多把锁了，则判断每把锁的兼容性，都兼容才能加锁，不兼容就得等待
                                boolean compatibility = true;
                                for (Lock lock : locks) {
                                    compatibility = LockSys.lock_compatibility_matrix[LockSys.getLockMode(lock.getMode())][LockSys.getLockMode(mode)];
                                    if (!compatibility) {
                                        // 阻塞等待，等待当前不兼容的锁释放掉，假如记录依次加了lock1、lock2，那么lock3与这两个都不兼容，那么就先等待lock1释放掉，释放掉之后，需要再此判断
                                        Lock waitLock = Lock.builder()
                                                .trx(TrxSys.trxThreadLocal.get())
                                                .mode(mode)
                                                .recId(recId)
                                                .isWait(true)
                                                .lock(new Object())
                                                .thread(Thread.currentThread())
                                                .build();
                                        LockSys.addLock(recId, waitLock);
                                        synchronized (waitLock.getLock()) {
                                            try {
                                                waitLock.getLock().wait();
                                                break;
                                            } catch (InterruptedException e) {
                                                throw new RuntimeException(e);
                                            }
                                        }
                                    }
                                }

                                // 都兼容，则直接加锁
                                if (compatibility) {
                                    LockSys.addLock(recId, Lock.builder()
                                            .trx(TrxSys.trxThreadLocal.get())
                                            .mode(mode)
                                            .recId(recId)
                                            .lock(new Object())
                                            .thread(Thread.currentThread())
                                            .build());
                                }
                            }
                        }

                        result.add(recContent);
                    }
                }

            }

            pageCur = selectOneResult.getPageCur();
        }

        if (orderByColumn != null) {
            String finalOrderByColumn = orderByColumn;
            boolean finalAsc = asc;
            CollUtil.sort(sortBuf, (o1, o2) -> {
                String s1 = (String) o1.get(finalOrderByColumn);
                String s2 = (String) o2.get(finalOrderByColumn);
                if (finalAsc) {
                    return CompareUtil.compare(s1, s2);
                } else {
                    return CompareUtil.compare(s2, s1);
                }
            });

            // result相当于客户端
            for (HashMap<String, Object> entry : sortBuf) {
                result.add(entry);
            }
        }

        // 如果是group by，则需要把heapTable中的数据进行返回
        if (groupByColumn != null) {
            result.add(heapTable);
        }

        return result;
    }

    /**
     * 加辅助索引的底层实现
     * @param alter 加索引的alter语句
     */
    public void alter(Alter alter) {
        Table table = alter.getTable();
        DictTable dictTable = SystemDict.getInstance().getNameTables().get(table.getName());

        List<AlterExpression> alterExpressions = alter.getAlterExpressions();
        Index index = alterExpressions.get(0).getIndex();

        DictIndex dictIndex = new DictIndex();
        dictIndex.setIndexName(index.getName());
        dictIndex.setIndexId(SystemDict.getInstance().addMaxIndexId());
        dictIndex.setTableId(dictTable.getTableId());

        // 一个索引定义了有哪些字段
        List<String> fieldNames = new ArrayList<>();
        for (Index.ColumnParams column : index.getColumns()) {
            String columnName = column.getColumnName();
            DictColumn dictColumn = dictTable.getDictColumnList().stream().filter(c -> c.getName().equals(columnName)).findFirst().get();
            DictField dictField = new DictField();
            dictField.setDictColumn(dictColumn);
            dictField.setPrefixLen(dictColumn.getLen());  // 写死，暂时不支持前缀索引
            dictIndex.getDictFieldList().add(dictField);
            fieldNames.add(columnName);
        }

        // 记录一个表有哪些索引
        List<DictIndex> dictIndices = SystemDict.getInstance().getTableIdIndexes().get(dictTable.getTableId());
        if (CollUtil.isEmpty(dictIndices)) {
            dictIndices = new ArrayList<>();
        }
        dictIndices.add(dictIndex);
        SystemDict.getInstance().getTableIdIndexes().put(dictTable.getTableId(), dictIndices);

        // 先在表空间中开辟新页作为索引的根页
        int rootIndexPageNo = PageUtil.createPage(dictTable.getSpaceId());
        dictIndex.setPageNo(rootIndexPageNo);
        IndexPage rootIndexPage = PageUtil.readPage(dictTable.getSpaceId(), rootIndexPageNo);

        // 持久化到系统字典中
        SystemDict.getInstance().serialize();


        // 再扫描聚集索引上的数据，将索引字段对应的数据取出来放入buf中
        List<BufEntry> buf = new ArrayList<>(); // buf本质上就是一块内存，List也是内存
        int bufMax = 4;
        List<File> tempFiles = new ArrayList<>();
        File tmpDir = FileUtil.file("tmp"); // 用来存临时文件的目录

        // 全表扫描
        DTuple emptyTuple = new DTuple();
        PageCur pageCur = new PageCur();
        pageCur.setPageNo(BtrUtil.getFirstLeafPage(dictTable).fil_page_get_page_offset());
        pageCur.setRec(PAGE_NEW_INFIMUM);
        while (true) {
            SelectOneResult selectOneResult = haInnodb.select_one(dictTable, pageCur, emptyTuple, true, false);
            boolean endOfIndex = selectOneResult.isEndOfIndex();

            if (endOfIndex) {
                break;
            }

            // 某条记录内容
            HashMap<String, Object> recContent = selectOneResult.getRecContent();

            if (recContent != null) {

                // 构造一个Entry，存入buf用来排序，后续再存到索引中，用fields进行排序
                BufEntry bufEntry = new BufEntry();
                bufEntry.setId((Integer) recContent.get("id"));
                for (String fieldName : fieldNames) {
                    bufEntry.getFields().add((String) recContent.get(fieldName));
                }
                buf.add(bufEntry);

                if (buf.size() >= bufMax) {
                    // 如果buf满，则先排序，并先存到临时文件中，并清空buf
                    // 需要利用bufEntry中对应的索引字段进行排序，需要比较大小，这里跟对应字段的大小规则有关系，我们只跟

                    CollUtil.sort(buf, (entry1, entry2) -> {

                        int size = entry1.getFields().size();

                        for (int i = 0; i < size; i++) {
                            // 如果两个字段不相等，则比较直接返回这两个字段的大小关系，否则继续比较第二个字段
                            if (!entry1.getFields().get(i).equals(entry2.getFields().get(i))) {
                                return entry1.getFields().get(i).compareTo(entry2.getFields().get(i));
                            }
                        }

                        // 所有字段都相等则返回0
                        return 0;
                    });

                    // 存到临时文件
                    // 得按自定义格式来存，不能直接把buf序列化到文件中，因为如果后面都记录List来进行归并排序，那么越往后List会越大，那不符合需求了
                    // 需求是一个List是有固定大小的，是由于一个List放不下所有数据才用文件排序，所以文件排序时不能再把所有数据又加回到内存中
                    // 只能按自定义格式一条一条数据写到文件，后面一条一条数据从文件读取出来进行排序
                    File tempFile = FileUtil.createTempFile(tmpDir);
                    for (BufEntry entry : buf) {
                        FileUtil.appendUtf8String(JSONUtil.toJsonStr(entry), tempFile);
                        FileUtil.appendUtf8String(LineSeparator.MAC.getValue(), tempFile);
                    }
                    tempFiles.add(tempFile);

                    // 清空buf
                    buf.clear();
                }
            }

            pageCur = selectOneResult.getPageCur();
        }

        // 数据都遍历完了，先检查前面是否生成了临时文件，如果生成了临时文件，则需要进行归并排序，并且把最终的排序结果要插入到索引中
        if (CollUtil.isNotEmpty(tempFiles)) {

            // 如果生成了临时文件，但是数据遍历完了之后，最后一个buf没有满，此时也需要把最后一个buf生成临时文件，一起进行归并排序
            if (CollUtil.isNotEmpty(buf)) {
                File tempFile = FileUtil.createTempFile(tmpDir);
                for (BufEntry entry : buf) {
                    FileUtil.appendUtf8String(JSONUtil.toJsonStr(entry), tempFile);
                    FileUtil.appendUtf8String(LineSeparator.MAC.getValue(), tempFile);
                }
                tempFiles.add(tempFile);

                // 清空buf
                buf.clear();
            }

            try {
                File sortResult = FileSort.sort(tmpDir);

                BufferedReader reader = FileUtil.getUtf8Reader(sortResult);
                String s1 = reader.readLine();
                while (s1 != null) {
                    BufEntry bufEntry = JSONUtil.toBean(s1, BufEntry.class);
                    // 将bufEntry插入到索引
                    DTuple dTuple = DTupleUtil.convert(bufEntry);
                    PageUtil.insertRow(rootIndexPage, dTuple);
                    s1 = reader.readLine();
                }

                sortResult.delete();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        } else {
            CollUtil.sort(buf, (entry1, entry2) -> {

                int size = entry1.getFields().size();

                for (int i = 0; i < size; i++) {
                    // 如果两个字段不相等，则比较直接返回这两个字段的大小关系，否则继续比较第二个字段
                    if (!entry1.getFields().get(i).equals(entry2.getFields().get(i))) {
                        return entry1.getFields().get(i).compareTo(entry2.getFields().get(i));
                    }
                }

                // 所有字段都相等则返回0
                return 0;
            });

            for (BufEntry bufEntry : buf) {
                // 将bufEntry插入到索引
                DTuple dTuple = DTupleUtil.convert(bufEntry);
                PageUtil.insertRow(rootIndexPage, dTuple);
            }
        }
    }

    /**
     * block nested-loop join查询的实现
     * @param selectStatement sql语句
     * @param joins join信息
     * @return join结果，会把list转成string
     */
    public String join(PlainSelect selectStatement, List<Join> joins) {
        // t1 join t2中的t1
        Table fromTable = (Table) selectStatement.getFromItem();
        String fromTableName = fromTable.getName();
        DictTable fromDictTable = SystemDict.getInstance().getNameTables().get(fromTableName);

        // t1 join t2中的t2
        Join join = joins.get(0);
        Table joinTable = (Table) join.getFromItem();
        String joinTableName = joinTable.getName();
        DictTable joinDictTable = SystemDict.getInstance().getNameTables().get(joinTableName);

        // join的on条件
        LinkedList<Expression> onExpressions = (LinkedList<Expression>) join.getOnExpressions();
        EqualsTo joinEqualsTo = (EqualsTo) onExpressions.get(0);
        Column joinLeftExpression = (Column) joinEqualsTo.getLeftExpression();
        Column joinRightExpression = (Column) joinEqualsTo.getRightExpression();
        String joinLeftTableName = joinLeftExpression.getTable().getName();
        String joinLeftColumnName = joinLeftExpression.getColumnName();
        String joinRightTableName = joinRightExpression.getTable().getName();
        String joinRightColumnName = joinRightExpression.getColumnName();


        // where条件
        EqualsTo whereEqualsTo = (EqualsTo) selectStatement.getWhere();
        Column whereLeftExpression = (Column) whereEqualsTo.getLeftExpression();
        StringValue whereRightExpression = (StringValue) whereEqualsTo.getRightExpression();
        String whereTableName = whereLeftExpression.getTable().getName();
        String whereColumnName = whereLeftExpression.getColumnName();
        String whereValue = whereRightExpression.getValue();

        // join之后的结果
        List<String> result = new ArrayList<>();

        List<HashMap<String, Object>> joinBuffer = new ArrayList<>();
        int maxJoinBuffer = 3;

        // 1. 遍历驱动表的内容，可以走a字段索引，也可以走全表扫描，这里只实现全表扫描
        DTuple emptyTuple = new DTuple();
        PageCur leftPageCur = new PageCur();
        leftPageCur.setPageNo(BtrUtil.getFirstLeafPage(fromDictTable).fil_page_get_page_offset());
        leftPageCur.setRec(PAGE_NEW_INFIMUM);
        while (true) {
            SelectOneResult leftSelectOneResult = haInnodb.select_one(fromDictTable, leftPageCur, emptyTuple, true, false);
            boolean leftEndOfIndex = leftSelectOneResult.isEndOfIndex();
            if (leftEndOfIndex) {
                break;
            }

            // 驱动表的某条记录内容
            HashMap<String, Object> leftRec = leftSelectOneResult.getRecContent();

            if (leftRec != null) {
                // where条件过滤
                if (whereTableName.equals(fromTableName) && whereValue.equals(leftRec.get(whereColumnName))) {
                    joinBuffer.add(leftRec);

                    // joinBuffer满了，则直接进行部分join
                    if (joinBuffer.size() >= maxJoinBuffer) {
                        PageCur rightPageCur = new PageCur();
                        rightPageCur.setPageNo(BtrUtil.getFirstLeafPage(joinDictTable).fil_page_get_page_offset());
                        rightPageCur.setRec(PAGE_NEW_INFIMUM);
                        while (true) {
                            SelectOneResult rightSelectOneResult = haInnodb.select_one(joinDictTable, rightPageCur, emptyTuple, true, false);
                            boolean rightEndOfIndex = rightSelectOneResult.isEndOfIndex();
                            if (rightEndOfIndex) {
                                break;
                            }

                            // 被驱动表的某条记录内容
                            HashMap<String, Object> rightRec = rightSelectOneResult.getRecContent();

                            for (HashMap<String, Object> leftRecBuffer : joinBuffer) {
                                // on条件过滤
                                if (leftRecBuffer.get(joinLeftColumnName).equals(rightRec.get(joinRightColumnName))) {
                                    HashMap<String, Object> resultHashMap = new HashMap<>();
                                    for (Map.Entry<String, Object> entry : leftRecBuffer.entrySet()) {
                                        resultHashMap.put(joinLeftTableName + "." + entry.getKey(), entry.getValue());
                                    }
                                    for (Map.Entry<String, Object> entry : rightRec.entrySet()) {
                                        resultHashMap.put(joinRightTableName + "." + entry.getKey(), entry.getValue());
                                    }
                                    result.add(resultHashMap.toString());
                                }
                            }

                            rightPageCur = rightSelectOneResult.getPageCur();
                        }

                        joinBuffer.clear();
                    }
                }
            }

            leftPageCur = leftSelectOneResult.getPageCur();

        }


        // joinBuffer有数据，要继续join
        if (CollUtil.isNotEmpty(joinBuffer)) {
            PageCur rightPageCur = new PageCur();
            rightPageCur.setPageNo(BtrUtil.getFirstLeafPage(joinDictTable).fil_page_get_page_offset());
            rightPageCur.setRec(PAGE_NEW_INFIMUM);
            while (true) {
                SelectOneResult rightSelectOneResult = haInnodb.select_one(joinDictTable, rightPageCur, emptyTuple, true, false);
                boolean rightEndOfIndex = rightSelectOneResult.isEndOfIndex();
                if (rightEndOfIndex) {
                    break;
                }

                // 被驱动表的某条记录内容
                HashMap<String, Object> rightRec = rightSelectOneResult.getRecContent();

                for (HashMap<String, Object> leftRecBuffer : joinBuffer) {
                    // on条件过滤
                    if (leftRecBuffer.get(joinLeftColumnName).equals(rightRec.get(joinRightColumnName))) {
                        HashMap<String, Object> resultHashMap = new HashMap<>();
                        for (Map.Entry<String, Object> entry : leftRecBuffer.entrySet()) {
                            resultHashMap.put(joinLeftTableName + "." + entry.getKey(), entry.getValue());
                        }
                        for (Map.Entry<String, Object> entry : rightRec.entrySet()) {
                            resultHashMap.put(joinRightTableName + "." + entry.getKey(), entry.getValue());
                        }
                        result.add(resultHashMap.toString());
                    }
                }

                rightPageCur = rightSelectOneResult.getPageCur();
            }

            joinBuffer.clear();
        }

        return result.toString();
    }
}
